///*
// * Copyright (C) 2010-2101 Alibaba Group Holding Limited.
// *
// * Licensed under the Apache License, Version 2.0 (the "License");
// * you may not use this file except in compliance with the License.
// * You may obtain a copy of the License at
// *
// *     http://www.apache.org/licenses/LICENSE-2.0
// *
// * Unless required by applicable law or agreed to in writing, software
// * distributed under the License is distributed on an "AS IS" BASIS,
// * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// * See the License for the specific language governing permissions and
// * limitations under the License.
// */
//
//package com.alibaba.otter.node.etl.load.loader.mq;
//
//import com.alibaba.otter.node.etl.load.exception.LoadException;
//import com.alibaba.otter.node.etl.load.loader.AbstractLoadAction;
//import com.alibaba.otter.node.etl.load.loader.db.DbLoadDumper;
//import com.alibaba.otter.node.etl.load.loader.db.DbLoadMerger;
//import com.alibaba.otter.node.etl.load.loader.db.context.DataLoadContext;
//import com.alibaba.otter.node.etl.load.loader.mq.fomatter.Formatter;
//import com.alibaba.otter.node.etl.load.loader.mq.producer.MQProducer;
//import com.alibaba.otter.node.etl.load.loader.mq.producer.MqProducerFactory;
//import com.alibaba.otter.shared.common.model.config.ConfigHelper;
//import com.alibaba.otter.shared.common.model.config.data.DataMedia;
//import com.alibaba.otter.shared.common.model.config.data.mq.MqDataMedia;
//import com.alibaba.otter.shared.etl.model.EventData;
//import com.google.common.collect.ImmutableMap;
//import org.apache.commons.lang.exception.ExceptionUtils;
//import org.springframework.dao.DeadlockLoserDataAccessException;
//import org.springframework.util.CollectionUtils;
//
//import java.sql.Statement;
//import java.util.ArrayList;
//import java.util.Collections;
//import java.util.List;
//import java.util.Map;
//import java.util.concurrent.Callable;
//
///**
// * MQ load的执行入口
// *
// * @author zhenglincheng
// * @version 0.0.1
// * @since 2021/5/11 19:17
// */
//public class MqLoadAction extends AbstractLoadAction {
//
//    private int retry = 3;
//
//    private int retryWait = 3000;
//
//    private final String workNameFormat;
//
//    private MqProducerFactory mqProducerFactory;
//
//    private Formatter formatter;
//
//
//    public MqLoadAction() {
//
//        this.workName = "MQLoadAction";
//        this.workNameFormat = "pipelineId = %s , pipelineName = %s , " + workName;
//    }
//
//
//
//    @Override
//    protected void load(DataLoadContext context, List<EventData> items) {
//
//        //执行前置流程
//        for (final EventData item : items) {
//            interceptor.before(context, item);
//        }
//        if (context.getPipeline().getParameters().isDryRun()) {
//            doDryRun(context, Collections.singletonList(items));
//            return;
//        }
//        //是否启用数据合并
//        if (context.getPipeline().getParameters().getEnableMerge()) {
//            // 进行一次数据合并，合并相同pk的多次I/U/D操作
//            items = DbLoadMerger.merge(items);
//        }
//        //判定是否需要进行dml数据缓存
//        if (CollectionUtils.isEmpty(items)){
//            return;
//        }
//        doTwoPhase(context, Collections.singletonList(items));
//
//    }
//
//    protected void doLoadFromReleaseDml(DataLoadContext context, List<EventData> loadData) {
//        if (context.getPipeline().getParameters().isDryRun()) {
//            doDryRun(context, Collections.singletonList(loadData));
//        } else {
//            doTwoPhase(context, Collections.singletonList(loadData));
//        }
//    }
//
//    @Override
//    protected Callable<Exception> getWorker(DataLoadContext context, List<EventData> eventDataCollection, boolean canBatch) {
//        return new MqLoadWorker(context, eventDataCollection, canBatch);
//    }
//
//    /**
//     * 执行ddl的调用，处理逻辑比较简单: 串行调用
//     *
//     * @param context
//     * @param data
//     */
//    @Override
//    protected boolean doLoadDDL(DataLoadContext context, EventData data) throws Exception{
//
//        MqDataMedia target = (MqDataMedia) ConfigHelper.findDataMedia(context.getPipeline(), data.getTableId());
//        ImmutableMap<String, Object> message = ImmutableMap.<String, Object>builder()
//                .put("type", data.getEventType())
//                .put("schema_name", data.getSchemaName())
//                .put("sql", data.getSql())
//                .build();
//        DataMedia<?> dataMedia = ConfigHelper.findDataMedia(context.getPipeline(), data.getTableId());
//        MQProducer mqProducer = mqProducerFactory.getMQProducer(context.getIdentity().getPipelineId(), (MqDataMedia) dataMedia);
//        return mqProducer.send(target.getPartition(), target.getKey(), target.getNamespace(), message);
//
//    }
//
//    protected void adjustConfig(DataLoadContext context) {
//
//    }
//
//
//    enum ExecuteResult {
//        SUCCESS, ERROR, RETRY
//    }
//
//    class MqLoadWorker implements Callable<Exception> {
//
//        private final boolean canBatch;
//        private final DataLoadContext context;
//        private final MQProducer mqProducer;
//        private final MqDataMedia mqDataMedia;
//        private final List<EventData> eventDataList;
//        private final List<EventData> allFailedDataList = new ArrayList<>();
//        private final List<EventData> allProcesedDataList = new ArrayList<>();
//        private final List<EventData> processedDataList = new ArrayList<>();
//        private final List<EventData> failedDataList = new ArrayList<>();
//
//        public MqLoadWorker(DataLoadContext context, List<EventData> eventDataList, boolean canBatch) {
//            this.context = context;
//            this.eventDataList = eventDataList;
//            this.canBatch = canBatch;
//            EventData data = eventDataList.get(0); // eventData为同一数据库的记录，只取第一条即可
//            this.mqDataMedia = (MqDataMedia) ConfigHelper.findDataMedia(context.getPipeline(), data.getTableId());
//            this.mqProducer = mqProducerFactory.getMQProducer(context.getIdentity().getPipelineId(), mqDataMedia);
//        }
//
//        public Exception call() throws Exception {
//            try {
//                Thread.currentThread().setName(String.format(workNameFormat,
//                        context.getPipeline().getId(),
//                        context.getPipeline().getName()));
//                return doCall();
//            } finally {
//                Thread.currentThread().setName(workName);
//            }
//        }
//
//        private Exception doCall() {
//            RuntimeException error;
//            ExecuteResult exeResult;
//            int index = 0;// 记录下处理成功的记录下标
//            while (index < eventDataList.size()) {
//                // 处理数据切分
//                final List<EventData> splitDatas = new ArrayList<>();
//                if (useBatch && canBatch) {
//                    int end = Math.min(index + batchSize, eventDataList.size());
//                    splitDatas.addAll(eventDataList.subList(index, end));
//                    index = end;// 移动到下一批次
//                } else {
//                    splitDatas.add(eventDataList.get(index));
//                    index = index + 1;// 移动到下一条
//                }
//
//                int retryCount = 0;
//
//
//                while (true) {
//                    try {
//                        if (!CollectionUtils.isEmpty(failedDataList)) {
//                            splitDatas.clear();
//                            splitDatas.addAll(failedDataList); // 下次重试时，只处理错误的记录
//                        } else {
//                            failedDataList.addAll(splitDatas); // 先添加为出错记录，可能获取lob,datasource会出错
//                        }
//
//                        if (useBatch && canBatch) {
//                            List<EventData> sendData = new ArrayList<>();
//                            List<Map<String,Object>> messages = new ArrayList<>();
//
//                            for (EventData eventData : splitDatas) {
//                                //mq发送数据格式化
//                                Map<String, Object> message = formatter.format(context, eventData);
//                                //格式化存在数据数据过滤，当返回数据为空时跳过数据发送
//                                if (message == null) {
//                                    continue;
//                                }
//                                sendData.add(eventData);
//                                messages.add(message);
//                            }
//                            //消息批量发送
//                            mqProducer.send(mqDataMedia.getPartition(), mqDataMedia.getKey(), mqDataMedia.getNamespace(), messages,(i,result) -> {
//                                // 更新统计信息
//                                processStat(sendData.get(i),result? 1:0,true);
//                            });
//                        } else {
//                            final EventData data = splitDatas.get(0);// 直接取第一条
//                            int affect = 0;
//                            //mq发送数据格式化
//                            Map<String, Object> message = formatter.format(context, data);
//                            //格式化存在数据数据过滤，当返回数据为空时跳过数据发送
//                            if (message == null) {
//                                break;
//                            }
//                            // 发送message到topic
//                            if (mqProducer.send(mqDataMedia.getPartition(), mqDataMedia.getKey(), mqDataMedia.getNamespace(), message)) {
//                                affect = 1;
//                            }
//                            // 更新统计信息
//                            processStat(data, affect, false);
//                        }
//
//                        error = null;
//                        exeResult = ExecuteResult.SUCCESS;
//                    } catch (DeadlockLoserDataAccessException ex) {
//                        error = new LoadException(ExceptionUtils.getFullStackTrace(ex),
//                                DbLoadDumper.dumpEventDatas(splitDatas));
//                        exeResult = ExecuteResult.RETRY;
//                    } catch (Throwable ex) {
//                        error = new LoadException(ExceptionUtils.getFullStackTrace(ex),
//                                DbLoadDumper.dumpEventDatas(splitDatas));
//                        // if (StringUtils.contains(ex.getMessage(),
//                        // "ORA-00001")) {
//                        // exeResult = ExecuteResult.RETRY;
//                        // } else {
//                        // exeResult = ExecuteResult.ERROR;
//                        // }
//                        exeResult = ExecuteResult.ERROR;
//                    }
//
//                    if (ExecuteResult.SUCCESS == exeResult) {
//                        allFailedDataList.addAll(failedDataList);// 记录一下异常到all记录中
//                        allProcesedDataList.addAll(processedDataList);
//                        failedDataList.clear();// 清空上一轮的处理
//                        processedDataList.clear();
//                        break; // do next eventData
//                    } else if (ExecuteResult.RETRY == exeResult) {
//                        retryCount = retryCount + 1;// 计数一次
//                        // 出现异常，理论上当前的批次都会失败
//                        processedDataList.clear();
//                        failedDataList.clear();
//                        failedDataList.addAll(splitDatas);
//                        if (retryCount >= retry) {
//                            processFailedDatas(index);// 重试已结束，添加出错记录并退出
//                            throw new LoadException(String.format("execute [%s] retry %s times failed",
//                                    context.getIdentity().toString(),
//                                    retryCount), error);
//                        } else {
//                            try {
//                                int wait = retryCount * retryWait;
//                                wait = Math.max(wait, retryWait);
//                                Thread.sleep(wait);
//                            } catch (InterruptedException ex) {
//                                Thread.interrupted();
//                                processFailedDatas(index);// 局部处理出错了
//                                throw new LoadException(ex);
//                            }
//                        }
//                    } else {
//                        // 出现异常，理论上当前的批次都会失败
//                        processedDataList.clear();
//                        failedDataList.clear();
//                        failedDataList.addAll(splitDatas);
//                        processFailedDatas(index);// 局部处理出错了
//                        throw error;
//                    }
//                }
//            }
//            // 记录一下当前处理过程中失败的记录,affect = 0的记录
//            context.getFailedDatas().addAll(allFailedDataList);
//            context.getProcessedDatas().addAll(allProcesedDataList);
//            return null;
//        }
//
//        private void processStat(EventData data, int affect, boolean batch) {
//            if (batch && (affect < 1 && affect != Statement.SUCCESS_NO_INFO)) {
//                failedDataList.add(data); // 记录到错误的临时队列，进行重试处理
//            } else if (!batch && affect < 1) {
//                failedDataList.add(data);// 记录到错误的临时队列，进行重试处理
//            } else {
//                processedDataList.add(data); // 记录到成功的临时队列，commit也可能会失败。所以这记录也可能需要进行重试
//                MqLoadAction.this.processStat(data, context);
//            }
//        }
//
//        // 出现异常回滚了，记录一下异常记录
//        private void processFailedDatas(int index) {
//            allFailedDataList.addAll(failedDataList);// 添加失败记录
//            context.getFailedDatas().addAll(allFailedDataList);// 添加历史出错记录
//            for (; index < eventDataList.size(); index++) { // 记录一下未处理的数据
//                context.getFailedDatas().add(eventDataList.get(index));
//            }
//            // 这里不需要添加当前成功记录，出现异常后会rollback所有的成功记录，比如processDatas有记录，但在commit出现失败
//            // (bugfix)
//            allProcesedDataList.addAll(processedDataList);
//            context.getProcessedDatas().addAll(allProcesedDataList);// 添加历史成功记录
//        }
//
//    }
//
//    // =============== setter / getter ===============
//
//    public void setRetry(int retry) {
//        this.retry = retry;
//    }
//
//    public void setRetryWait(int retryWait) {
//        this.retryWait = retryWait;
//    }
//
//
//    public void setMqProducerFactory(MqProducerFactory mqProducerFactory) {
//        this.mqProducerFactory = mqProducerFactory;
//    }
//
//    public void setFormatter(Formatter formatter) {
//        this.formatter = formatter;
//    }
//}
